#pragma once
#include "requestor.hpp"
#include <unordered_set>

//消息发布客户端:创建主题,删除主题,发布消息
//消息订阅客户端:创建主题,删除主题,订阅主题,取消订阅
//五个操作接口->创建/删除/订阅/取消订阅/发布
//消息处理接口->dispatcher模块进行消息分发处理(把消息推送给所有订阅的客户端)
//内部数据管理:主题名称与消息处理回调函数的映射关系->订阅客户端什么时候收到消息是不确定的,当发布客户端发布了消息,订阅客户端才会收到推送过来的消息
//订阅客户端无法进行消息的同步处理,只能使用异步进行消息处理,设置好消息的回调函数,收到消息直接调用消息的回调函数
namespace rpc
{
    namespace client
    {
        class TopicManager
        {
        public:
            //回调处理函数->订阅的回调函数类型:主题名称,消息
            using SubCallback = std::function<void(const std::string & key,const std::string & msg)>;
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager(const Requestor::ptr & requestor):_requestor(requestor){}

            //创建主题
            bool create(const BaseConnection::ptr & conn,const std::string & key)
            {
                return commonRequest(conn,key,TopicOptype::TOPIC_CREATE);
            }

            //删除主题
            bool remove(const BaseConnection::ptr & conn,const std::string & key)
            {
                return commonRequest(conn,key,TopicOptype::TOPIC_REMOVE);
            }

            //订阅主题
            bool subscribe(const BaseConnection::ptr & conn,const std::string & key,const SubCallback & cb)
            {
                addSubscribe(key,cb);
                bool ret = commonRequest(conn,key,TopicOptype::TOPIC_SUBSCRIBE);
                if(ret == false)
                {
                    delSubscribe(key);//订阅失败,删除这个对象
                    return false;
                }
                return true;
            }

            //取消订阅
            bool cancel(const BaseConnection::ptr & conn,const std::string & key)
            {
                delSubscribe(key);
                return commonRequest(conn,key,TopicOptype::TOPIC_CANCEL);
            }

            //发布消息
            bool publish(const BaseConnection::ptr & conn,const std::string & key,const std::string &msg)
            {
                return commonRequest(conn,key,TopicOptype::TOPIC_PUBLISH,msg);
            }

            //提供给dispatcher模块,消息发送客户端发送消息之后,服务端主动广播的消息
            void onPublish(const BaseConnection::ptr & conn,const TopicRequest::ptr & msg)
            {
                //从消息中取出操作类型->是否是消息推送请求
                //取出主题名称以及消息内容
                //通过主题名称查找对应的回调处理函数,有则调用,无则出错
                auto type = msg->optype();
                if(type != TopicOptype::TOPIC_PUBLISH)
                {
                    LOG(ERROR,"收到了错误类型的主题操作");
                    return;
                }
                std::string topic_key = msg->topicKey();
                std::string topic_msg = msg->topicMsg();
                auto callback = getSubscribe(topic_key);
                if(!callback) //空的函数对象
                {
                    LOG(ERROR,"收到了 %s 主题消息,但是该消息无主题处理回调",topic_key.c_str());
                    return;
                }
                return callback(topic_key,topic_msg);
            }

        private:
            //添加映射关系
            void addSubscribe(const std::string & key,const SubCallback & cb)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.insert(std::make_pair(key,cb));
            }
            
            //删除这个主题
            void delSubscribe(const std::string & key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                _topic_callbacks.erase(key);
            }

            //获取主题回调函数
            const SubCallback getSubscribe(const std::string & key)
            {
                std::unique_lock<std::mutex> lock(_mutex);
                auto it = _topic_callbacks.find(key);
                if(it == _topic_callbacks.end())
                {
                    return SubCallback();
                }
                return it->second;
            }

            //通用的Request请求
            bool commonRequest(const BaseConnection::ptr & conn,const std::string & key,TopicOptype type,const std::string & msg = "")
            {
                //构造请求对象,填充数据
                //向服务端发送请求,等待响应
                //判断请求处理是否成功
                auto msg_req = MessageFactory::create<TopicRequest>();
                msg_req->setId(UUID::uuid());
                msg_req->setMType(MType::REQ_TOPIC);
                msg_req->setOptype(type);
                msg_req->setTopicKey(key);
                if(type == TopicOptype::TOPIC_PUBLISH)//发布主题
                {
                    msg_req->setTopicMsg(msg);
                }
                BaseMessage::ptr msg_rsp;
                bool ret = _requestor->send(conn,msg_req,msg_rsp);
                if(ret == false)
                {
                    LOG(ERROR,"主题操作请求失败");
                    return false;
                }
                auto topic_rsp_msg = std::dynamic_pointer_cast<TopicResponse>(msg_rsp);
                if(!topic_rsp_msg)
                {
                    LOG(ERROR,"主题操作响应,向下类型转换失败");
                    return false;
                }
                if(topic_rsp_msg->rcode() != RCode::RCODE_OK)
                {
                    LOG(ERROR,"主题操作请求出错:%s",errReason(topic_rsp_msg->rcode()));
                    return false;
                }
                return true;
            }
        private:
            std::mutex _mutex;
            std::unordered_map<std::string,SubCallback> _topic_callbacks;//主题名称:回调函数
            Requestor::ptr _requestor;//向服务端发送请求
        };
    }
}